package com.gugee.ins.data.common.consumer;

import com.gugee.common.exception.BizException;
import com.gugee.common.exception.SpiderRetryableException;
import com.gugee.ins.data.common.msg.InsMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.client.ResourceAccessException;

import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * Created with IntelliJ IDEA.
 * Description:消息处理器,采用线程池处理，
 * User: luolifeng
 * Date: 2021-01-07 11:22
 */
@Slf4j
public abstract class AbstractGugeeMessageHandler<T> {
    //线程池
    private final ThreadPoolExecutor customPoolMain;
    private final String  moduleName;

    public AbstractGugeeMessageHandler(ThreadPoolExecutor customPoolMain) {
        this.customPoolMain = customPoolMain;
        moduleName=this.getClass().getName();
    }
    //单个消息处理失败后的逻辑
    public abstract void failHandle(Exception e,T msg);
    //单个消息处理成功后的逻辑
    public abstract void successHandle(T msg);
    //单个消息处理逻辑
    public abstract void handleMsg(T message);
    public void handle(List<T> messages){

        Collection<Future<?>> futures = new LinkedList<>();
        messages.forEach(
                message -> futures.add(customPoolMain.submit(() -> {

                    try {
                        handleMsg(message);
                        successHandle(message);
                    } catch(Exception e){
                        failHandle(e,message);
                    }

                }))
        );
        for (Future<?> future : futures) {
            try {
                future.get();
            } catch(Exception  e) {
                e.printStackTrace();
                log.error("{}: future .{}" ,moduleName,e.getMessage());
            }
        }
    }

}
